package jupitermouse.site.source.custom.mysql;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.sql.*;

/**
 * @desc 自定义SQL数据源
 * @author renqiqiang
 */
@Slf4j
public class MySqlDataSourceFunction extends RichSourceFunction<String> {

    /**
     * 运行状态
     */
    private volatile Boolean running = true;

    /**
     * 预执行
     */
    private PreparedStatement ps;

    /**
     * 连接
     */
    private Connection connection;

    /**
     * open()方法建立连接
     * @param conf 配置参数
     * @throws Exception
     */
    @Override
    public void open(Configuration conf) throws Exception {
        super.open(conf);
        log.info("start connection...");
        String driverClass = conf.getString("custom.rdbms.driverClass", "com.mysql.jdbc.Driver");
        String url = conf.getString("custom.rdbms.url", "jdbc:mysql://127.0.0.1:3306/hdsp_platform?useSSL=false");
        String username = conf.getString("custom.rdbms.username", "root");
        String password = conf.getString("custom.rdbms.password", "root");

        connection = getConnection(driverClass, url, username, password);
        ps  = connection.prepareStatement("select code from iam_menu");
        log.info("connection end ...");
    }

    /**
     * 启动一个 source，即对接一个外部数据源然后 emit 元素形成 stream
     * @param ctx SourceContext
     * @throws Exception
     */
    @Override
    public void run(SourceContext ctx) throws Exception {
        log.info("run excute ...");
        while (running){
            ResultSet resultSet = ps.executeQuery();
            while (resultSet.next()){
                ctx.collect(resultSet.getString("code"));
            }
            Thread.sleep(1000);
        }
    }

    /**
     * 取消一个 source，也即将 run 中的循环 emit 元素的行为终止。
     */
    @Override
    public void cancel() {
        running = false;
    }

    /**
     * 获取数据源连接
     * @param driverClass 驱动类地址
     * @param url url
     * @param username 用户名
     * @param password 密码
     * @return Connection
     */
    private static Connection getConnection(String driverClass, String url, String username, String password){
        Connection connection = null;
        try {
            Class.forName(driverClass);
        } catch (Exception e) {
            log.info("driverClass:{} error", driverClass);
        }
        try {
            connection = DriverManager.getConnection(url, username, password);
        } catch (SQLException e) {
            log.info("connection: error \n url:{}\n username:{}\n password:{}\n", url, username, password);
        }
        return connection;
    }
}
